package com.iflytek.iot.redis.sharding.core;

import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;

/**
 * @author ZuRun
 * @date 2020/2/27 23:29:17
 */
public class RedisVirtualSlotsContainer {

    private final static String PREFIX = "##";
    /**
     * 虚拟节点与hash值对应关系
     */
    private final TreeMap<Long, String> virtualNodes = new TreeMap<>();
    private final Set<Long> slotHashCollections = new HashSet<>();


    public void addSlot(String slotName, int slotNodeNum) {
        synchronized (this) {
            for (int i = 0; i < slotNodeNum; i++) {
                String virtualNode = slotName + PREFIX + i;
                long hash = hash(virtualNode);
                if (virtualNodes.containsKey(hash)) {
                    // todo 也有可能是hash冲突
                    throw new RuntimeException("pls check slot repetitive, virtualNode: " + virtualNode);
                }
                virtualNodes.put(hash, virtualNode);
                slotHashCollections.add(hash);
            }
        }
    }

    public String findSlot(String key) {
        long hash = hash(key);
        final String slotName;
        // 如果hash值巧合已有，直接使用此key,否则使用小于此hash值的最近一个key
        if (slotHashCollections.contains(hash)) {
            slotName = virtualNodes.get(hash);
        } else {
            Map.Entry<Long, String> longStringEntry = virtualNodes.lowerEntry(hash);
            if (longStringEntry == null) {
                // 碰巧是最小的hash值，取第一个
                longStringEntry = virtualNodes.firstEntry();
            }
            slotName = longStringEntry.getValue();
        }
        return slotName.substring(0, slotName.indexOf(PREFIX));
    }

    /**
     * murmur hash算法实现
     */
    private long hash(byte[] key) {

        ByteBuffer buf = ByteBuffer.wrap(key);
        int seed = 0x1234ABCD;

        ByteOrder byteOrder = buf.order();
        buf.order(ByteOrder.LITTLE_ENDIAN);

        long m = 0xc6a4a7935bd1e995L;
        int r = 47;

        long h = seed ^ (buf.remaining() * m);

        long k;
        while (buf.remaining() >= 8) {
            k = buf.getLong();

            k *= m;
            k ^= k >>> r;
            k *= m;

            h ^= k;
            h *= m;
        }

        if (buf.remaining() > 0) {
            ByteBuffer finish = ByteBuffer.allocate(8).order(
                    ByteOrder.LITTLE_ENDIAN);
            // for big-endian version, do this first:
            // finish.position(8-buf.remaining());
            finish.put(buf).rewind();
            h ^= finish.getLong();
            h *= m;
        }

        h ^= h >>> r;
        h *= m;
        h ^= h >>> r;

        buf.order(byteOrder);
        return h;
    }

    private long hash(String key) {
        return hash(key.getBytes());
    }
}
